class BaseEtl:
    """
    消费者基类，定义消费者的基本结构和处理流程
    """
    pipelines = []  # 管道列表，用于存储数据处理的各个步骤


    def run(self):
        """
        执行 ETL 主流程
        1. 从数据源读取数据
        2. 对数据进行转换处理
        3. 将转换后的数据写入目标位置
        """
        read_data = self.read()  # 读取数据
        transform_data = self.transform(read_data)  # 数据转换
        self.write(transform_data)  # 数据写入

    def transform(self, transform_data):
        """
        对数据应用管道处理
        遍历 pipelines 列表，将数据依次通过每个管道进行处理
        """
        for pipeline in self.pipelines:
            transform_data = pipeline.apply(transform_data)  # 将数据传入当前管道进行处理
        return transform_data

    def read(self):
        """
        从数据源读取数据
        该方法需由子类实现
        """
        raise NotImplementedError

    def write(self, transform_data):
        """
        写入处理后的数据
        该方法需由子类实现
        """
        raise NotImplementedError
